package com.atguigu.gmall.realtime.bootstrap

import java.util

import com.alibaba.fastjson.JSONObject
import com.atguigu.gmall.realtime.util.{MyKafkaSink, MySQLUtil}
import redis.clients.jedis.Jedis
import scala.collection.JavaConverters._

object BootStrapInit {



  //1 读取mysql
  //2 把读取到的数据 按照canal的格式进行转换
  //3 发送kafka
  def main(args: Array[String]): Unit = {
        val jedis = new Jedis("hdp1",6379)
        val dimTables = jedis.smembers("dim_tables")
    for (dimTableName <- dimTables.asScala) {
      val sql = s"select * from $dimTableName"
      println(sql)
      //1 读取mysql
      val dimDataList: util.List[JSONObject] = MySQLUtil.queryList(sql)
      //2 把读取到的数据 按照canal的格式进行转换

      for (dimDataMysql <- dimDataList.asScala ) {
        val dimDataJsonObj = new JSONObject()
        dimDataJsonObj.put("data",Array(dimDataMysql))
        dimDataJsonObj.put("pkNames",Array("id"))
        dimDataJsonObj.put("type","INSERT")
        dimDataJsonObj.put("database","gmall0609")
        dimDataJsonObj.put("table",dimTableName)
        println(dimDataJsonObj.toJSONString)
        val key: String = dimDataMysql.getString("id")
        //3 发送kafka
        MyKafkaSink.send("ODS_BASE_DB_C",key,dimDataJsonObj.toJSONString )
      }





    }



  }

}
